/*-------------------------------------------------------------------------
 *
 * mac.c
 *	  PostgreSQL 类型定义用于 6 字节、EUI-48、MAC 地址。
 *
 * Portions Copyright (c) 1998-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *		  src/backend/utils/adt/mac.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "common/hashfn.h"
#include "lib/hyperloglog.h"
#include "libpq/pqformat.h"
#include "port/pg_bswap.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/inet.h"
#include "utils/sortsupport.h"


/*
 *	用于排序和比较的工具宏：
 */

#define hibits(addr) \
  ((unsigned long)(((addr)->a<<16)|((addr)->b<<8)|((addr)->c)))

#define lobits(addr) \
  ((unsigned long)(((addr)->d<<16)|((addr)->e<<8)|((addr)->f)))

/* macaddr 的排序支持 */
typedef struct
{
	int64		input_count;	/* 看到的非空值数量 */
	bool		estimating;		/* 如果估计基数为真 */

	hyperLogLogState abbr_card; /* 基数估算器 */
} macaddr_sortsupport_state;

static int	fc_macaddr_cmp_internal(macaddr *fc_a1, macaddr *fc_a2);
static int	fc_macaddr_fast_cmp(Datum fc_x, Datum fc_y, SortSupport fc_ssup);
static bool fc_macaddr_abbrev_abort(int fc_memtupcount, SortSupport fc_ssup);
static Datum fc_macaddr_abbrev_convert(Datum fc_original, SortSupport fc_ssup);

/*
 *	MAC 地址读取器。接受几种常见的表示法。
 */

Datum macaddr_in(PG_FUNCTION_ARGS)
{
	char	   *fc_str = PG_GETARG_CSTRING(0);
	macaddr    *fc_result;
#ifdef FDD//cppcheck
	unsigned int		fc_a,
#else
	int			fc_a,
#endif
				fc_b,
				fc_c,
				fc_d,
				fc_e,
				fc_f;
	char		fc_junk[2];
	int			fc_count;

	/* %1s 仅在尾部有非空格垃圾时匹配 */

	fc_count = sscanf(fc_str, "%x:%x:%x:%x:%x:%x%1s",
				   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%x-%x-%x-%x-%x-%x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%2x%2x%2x:%2x%2x%2x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%2x%2x%2x-%2x%2x%2x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%2x%2x.%2x%2x.%2x%2x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%2x%2x-%2x%2x-%2x%2x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		fc_count = sscanf(fc_str, "%2x%2x%2x%2x%2x%2x%1s",
					   &fc_a, &fc_b, &fc_c, &fc_d, &fc_e, &fc_f, fc_junk);
	if (fc_count != 6)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
				 errmsg("invalid input syntax for type %s: \"%s\"", "macaddr",
						fc_str)));

	if ((fc_a < 0) || (fc_a > 255) || (fc_b < 0) || (fc_b > 255) ||
		(fc_c < 0) || (fc_c > 255) || (fc_d < 0) || (fc_d > 255) ||
		(fc_e < 0) || (fc_e > 255) || (fc_f < 0) || (fc_f > 255))
		ereport(ERROR,
				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
				 errmsg("invalid octet value in \"macaddr\" value: \"%s\"", fc_str)));

	fc_result = (macaddr *) palloc(sizeof(macaddr));

	fc_result->a = fc_a;
	fc_result->b = fc_b;
	fc_result->c = fc_c;
	fc_result->d = fc_d;
	fc_result->e = fc_e;
	fc_result->f = fc_f;

	PG_RETURN_MACADDR_P(fc_result);
}

/*
 *	MAC 地址输出函数。固定格式。
 */

Datum macaddr_out(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr = PG_GETARG_MACADDR_P(0);
	char	   *fc_result;

	fc_result = (char *) palloc(32);

	snprintf(fc_result, 32, "%02x:%02x:%02x:%02x:%02x:%02x",
			 fc_addr->a, fc_addr->b, fc_addr->c, fc_addr->d, fc_addr->e, fc_addr->f);

	PG_RETURN_CSTRING(fc_result);
}

/*
 *		macaddr_recv			- 将外部二进制格式转换为 macaddr
 *
 * 外部表示法仅为六个字节，MSB 优先。
 */
Datum macaddr_recv(PG_FUNCTION_ARGS)
{
	StringInfo	fc_buf = (StringInfo) PG_GETARG_POINTER(0);
	macaddr    *fc_addr;

	fc_addr = (macaddr *) palloc(sizeof(macaddr));

	fc_addr->a = pq_getmsgbyte(fc_buf);
	fc_addr->b = pq_getmsgbyte(fc_buf);
	fc_addr->c = pq_getmsgbyte(fc_buf);
	fc_addr->d = pq_getmsgbyte(fc_buf);
	fc_addr->e = pq_getmsgbyte(fc_buf);
	fc_addr->f = pq_getmsgbyte(fc_buf);

	PG_RETURN_MACADDR_P(fc_addr);
}

/*
 *		macaddr_send			- 将 macaddr 转换为二进制格式
 */
Datum macaddr_send(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr = PG_GETARG_MACADDR_P(0);
	StringInfoData fc_buf;

	pq_begintypsend(&fc_buf);
	pq_sendbyte(&fc_buf, fc_addr->a);
	pq_sendbyte(&fc_buf, fc_addr->b);
	pq_sendbyte(&fc_buf, fc_addr->c);
	pq_sendbyte(&fc_buf, fc_addr->d);
	pq_sendbyte(&fc_buf, fc_addr->e);
	pq_sendbyte(&fc_buf, fc_addr->f);
	PG_RETURN_BYTEA_P(pq_endtypsend(&fc_buf));
}


/*
 *	排序比较函数：
 */

static int fc_macaddr_cmp_internal(macaddr *fc_a1, macaddr *fc_a2)
{
	if (hibits(fc_a1) < hibits(fc_a2))
		return -1;
	else if (hibits(fc_a1) > hibits(fc_a2))
		return 1;
	else if (lobits(fc_a1) < lobits(fc_a2))
		return -1;
	else if (lobits(fc_a1) > lobits(fc_a2))
		return 1;
	else
		return 0;
}

Datum macaddr_cmp(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_INT32(fc_macaddr_cmp_internal(fc_a1, fc_a2));
}

/*
 *	布尔比较。
 */

Datum macaddr_lt(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) < 0);
}

Datum macaddr_le(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) <= 0);
}

Datum macaddr_eq(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) == 0);
}

Datum macaddr_ge(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) >= 0);
}

Datum macaddr_gt(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) > 0);
}

Datum macaddr_ne(PG_FUNCTION_ARGS)
{
	macaddr    *fc_a1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_a2 = PG_GETARG_MACADDR_P(1);

	PG_RETURN_BOOL(fc_macaddr_cmp_internal(fc_a1, fc_a2) != 0);
}

/*
 * 支持 macaddr 的哈希索引的函数。
 */
Datum hashmacaddr(PG_FUNCTION_ARGS)
{
	macaddr    *fc_key = PG_GETARG_MACADDR_P(0);

	return hash_any((unsigned char *) fc_key, sizeof(macaddr));
}

Datum hashmacaddrextended(PG_FUNCTION_ARGS)
{
	macaddr    *fc_key = PG_GETARG_MACADDR_P(0);

	return hash_any_extended((unsigned char *) fc_key, sizeof(macaddr),
							 PG_GETARG_INT64(1));
}

/*
 * 算术函数：位运算非、与、或。
 */
Datum macaddr_not(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_result;

	fc_result = (macaddr *) palloc(sizeof(macaddr));
	fc_result->a = ~fc_addr->a;
	fc_result->b = ~fc_addr->b;
	fc_result->c = ~fc_addr->c;
	fc_result->d = ~fc_addr->d;
	fc_result->e = ~fc_addr->e;
	fc_result->f = ~fc_addr->f;
	PG_RETURN_MACADDR_P(fc_result);
}

Datum macaddr_and(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_addr2 = PG_GETARG_MACADDR_P(1);
	macaddr    *fc_result;

	fc_result = (macaddr *) palloc(sizeof(macaddr));
	fc_result->a = fc_addr1->a & fc_addr2->a;
	fc_result->b = fc_addr1->b & fc_addr2->b;
	fc_result->c = fc_addr1->c & fc_addr2->c;
	fc_result->d = fc_addr1->d & fc_addr2->d;
	fc_result->e = fc_addr1->e & fc_addr2->e;
	fc_result->f = fc_addr1->f & fc_addr2->f;
	PG_RETURN_MACADDR_P(fc_result);
}

Datum macaddr_or(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr1 = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_addr2 = PG_GETARG_MACADDR_P(1);
	macaddr    *fc_result;

	fc_result = (macaddr *) palloc(sizeof(macaddr));
	fc_result->a = fc_addr1->a | fc_addr2->a;
	fc_result->b = fc_addr1->b | fc_addr2->b;
	fc_result->c = fc_addr1->c | fc_addr2->c;
	fc_result->d = fc_addr1->d | fc_addr2->d;
	fc_result->e = fc_addr1->e | fc_addr2->e;
	fc_result->f = fc_addr1->f | fc_addr2->f;
	PG_RETURN_MACADDR_P(fc_result);
}

/*
 *	截断函数以允许比较 mac 制造商。
 *	根据 Alex Pilosov 的建议 <alex@pilosoft.com>
 */
Datum macaddr_trunc(PG_FUNCTION_ARGS)
{
	macaddr    *fc_addr = PG_GETARG_MACADDR_P(0);
	macaddr    *fc_result;

	fc_result = (macaddr *) palloc(sizeof(macaddr));

	fc_result->a = fc_addr->a;
	fc_result->b = fc_addr->b;
	fc_result->c = fc_addr->c;
	fc_result->d = 0;
	fc_result->e = 0;
	fc_result->f = 0;

	PG_RETURN_MACADDR_P(fc_result);
}

/*
 * SortSupport 策略函数。填充 SortSupport 结构体，提供
 * 通过缩略键进行比较所需的信息。
 */
Datum macaddr_sortsupport(PG_FUNCTION_ARGS)
{
	SortSupport fc_ssup = (SortSupport) PG_GETARG_POINTER(0);

	fc_ssup->comparator = fc_macaddr_fast_cmp;
	fc_ssup->ssup_extra = NULL;

	if (fc_ssup->abbreviate)
	{
		macaddr_sortsupport_state *fc_uss;
		MemoryContext fc_oldcontext;

		fc_oldcontext = MemoryContextSwitchTo(fc_ssup->ssup_cxt);

		fc_uss = palloc(sizeof(macaddr_sortsupport_state));
		fc_uss->input_count = 0;
		fc_uss->estimating = true;
		initHyperLogLog(&fc_uss->abbr_card, 10);

		fc_ssup->ssup_extra = fc_uss;

		fc_ssup->comparator = ssup_datum_unsigned_cmp;
		fc_ssup->abbrev_converter = fc_macaddr_abbrev_convert;
		fc_ssup->abbrev_abort = fc_macaddr_abbrev_abort;
		fc_ssup->abbrev_full_comparator = fc_macaddr_fast_cmp;

		MemoryContextSwitchTo(fc_oldcontext);
	}

	PG_RETURN_VOID();
}

/*
 * SortSupport "传统" 比较函数。从堆中抽取两个 MAC 地址并
 * 对其进行标准比较。
 */
static int fc_macaddr_fast_cmp(Datum fc_x, Datum fc_y, SortSupport fc_ssup)
{
	macaddr    *fc_arg1 = DatumGetMacaddrP(fc_x);
	macaddr    *fc_arg2 = DatumGetMacaddrP(fc_y);

	return fc_macaddr_cmp_internal(fc_arg1, fc_arg2);
}

/*
 * 估计缩略键优化有效性的回调。
 *
 * 我们不关注非缩略数据的基数，因为在权威的 macaddr 比较器中
 * 没有快速路径进行相等性比较。
 */
static bool fc_macaddr_abbrev_abort(int fc_memtupcount, SortSupport fc_ssup)
{
	macaddr_sortsupport_state *fc_uss = fc_ssup->ssup_extra;
	double		fc_abbr_card;

	if (fc_memtupcount < 10000 || fc_uss->input_count < 10000 || !fc_uss->estimating)
		return false;

	fc_abbr_card = estimateHyperLogLog(&fc_uss->abbr_card);

	/*
	 * 如果我们有超过 100k 的不同值，那么即使我们对许多
	 * 亿行进行排序，我们仍然可能会持平，撤销那么多缩略行的罚款
	 * 可能也不值得。在这一点上，我们停止计数，因为我们知道
	 * 我们现在已经完全承诺。
	 */
	if (fc_abbr_card > 100000.0)
	{
#ifdef TRACE_SORT
		if (trace_sort)
			elog(LOG,
				 "macaddr_abbrev: estimation ends at cardinality %f"
				 " after " INT64_FORMAT " values (%d rows)",
				 fc_abbr_card, fc_uss->input_count, fc_memtupcount);
#endif
		fc_uss->estimating = false;
		return false;
	}

	/*
	 * 目标最小基数为每 ~2k 的非空输入 1。0.5 行的
	 * 允许因子使我们能够在真正病态的数据上更早中止，
	 * 在前 2k（非空）行中只有一个缩略值。
	 */
	if (fc_abbr_card < fc_uss->input_count / 2000.0 + 0.5)
	{
#ifdef TRACE_SORT
		if (trace_sort)
			elog(LOG,
				 "macaddr_abbrev: aborting abbreviation at cardinality %f"
				 " below threshold %f after " INT64_FORMAT " values (%d rows)",
				 fc_abbr_card, fc_uss->input_count / 2000.0 + 0.5, fc_uss->input_count,
				 fc_memtupcount);
#endif
		return true;
	}

#ifdef TRACE_SORT
	if (trace_sort)
		elog(LOG,
			 "macaddr_abbrev: cardinality %f after " INT64_FORMAT
			 " values (%d rows)", fc_abbr_card, fc_uss->input_count, fc_memtupcount);
#endif

	return false;
}

/*
 * SortSupport 转换例程。将原始 macaddr 表示法
 * 转换为缩略键表示法。
 *
 * 将 6 字节 MAC 地址的字节打包到 Datum 中，并将其视为一个
 * 无符号整数以便进行比较。在 64 位机器上，末尾将有两个零填充字节。
 * 整数将转换为本地字节顺序以方便比较。
 */
static Datum fc_macaddr_abbrev_convert(Datum fc_original, SortSupport fc_ssup)
{
	macaddr_sortsupport_state *fc_uss = fc_ssup->ssup_extra;
	macaddr    *fc_authoritative = DatumGetMacaddrP(fc_original);
	Datum		fc_res;

	/*
	 * 在 64 位机器上，将 8 字节的 datum 清零并复制 6 字节的
	 * MAC 地址。低位字节的末尾将有两个填充零字节。
	 */
#if SIZEOF_DATUM == 8
	memset(&fc_res, 0, SIZEOF_DATUM);
	memcpy(&fc_res, fc_authoritative, sizeof(macaddr));
#else							/* SIZEOF_DATUM != 8 */
	memcpy(&res, authoritative, SIZEOF_DATUM);
#endif
	fc_uss->input_count += 1;

	/*
	 * 基数估计。估计使用 uint32，因此在 64 位架构上，
	 * 将两个 32 位部分进行异或操作以产生稍微多一点的熵。
	 * 这两个零字节对这个操作没有任何实际影响。
	 */
	if (fc_uss->estimating)
	{
		uint32		fc_tmp;

#if SIZEOF_DATUM == 8
		fc_tmp = (uint32) fc_res ^ (uint32) ((uint64) fc_res >> 32);
#else							/* SIZEOF_DATUM != 8 */
		tmp = (uint32) res;
#endif

		addHyperLogLog(&fc_uss->abbr_card, DatumGetUInt32(hash_uint32(fc_tmp)));
	}

	/*
	 * 在小端机器上进行字节交换。
	 *
	 * 这是必要的，以便 ssup_datum_unsigned_cmp()（一个无符号整数
	 * 三向比较器）在所有平台上正常工作。否则，
	 * 比较器必须使用一对指针调用 memcmp()，分别指向每个缩略键的
	 * 第一个字节，这样会更慢。
	 */
	fc_res = DatumBigEndianToNative(fc_res);

	return fc_res;
}
